package com.lhd.app.job;

import com.lhd.app.processor.PetTypeLabelProcessor;
import com.lhd.app.sink.LocalFileSink;
import com.lhd.app.bean.PetTypeLabel;
import com.lhd.app.bean.UserBehavior;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;

import java.util.*;

public class PetTypeLabelJob {
    
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        
        // 创建测试数据源
        DataStream<UserBehavior> behaviorStream = createTestDataStream(env);
        
        // 处理宠物类型标签
        DataStream<PetTypeLabel> petTypeLabels = behaviorStream
            .filter(PetTypeLabelProcessor::isPetRelatedBehavior)
            .keyBy(UserBehavior::getUserId)
            .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
            .process(new PetTypeLabelProcessor())
            .name("pet-type-label-processor");
        
        // 输出到本地文件
        petTypeLabels.addSink(new LocalFileSink("data/pet_type_labels.csv"))
            .name("local-file-sink-pet-type");
        
        env.execute("Pet Type Label Generation Job - Local Output");
    }
    
    // 创建测试数据流
    private static DataStream<UserBehavior> createTestDataStream(StreamExecutionEnvironment env) {
        List<UserBehavior> testData = Arrays.asList(
            // 用户1：养狗用户
            createBehavior("user001", "buy", "狗狗", "狗零食", "狗粮", System.currentTimeMillis() - 86400000L),
            createBehavior("user001", "cart", "狗狗", "狗玩具", "狗玩具", System.currentTimeMillis() - 172800000L),
            createBehavior("user001", "browse", "猫/狗玩具", "通用玩具", "猫狗通用玩具", System.currentTimeMillis() - 259200000L),
            
            // 用户2：养猫用户
            createBehavior("user002", "buy", "猫咪", "猫主粮", "猫粮", System.currentTimeMillis() - 86400000L),
            createBehavior("user002", "favorite", "猫咪", "猫零食", "猫零食", System.currentTimeMillis() - 172800000L),
            createBehavior("user002", "browse", "猫/狗保健品", "通用保健品", "猫狗通用保健品", System.currentTimeMillis() - 259200000L),
            
            // 用户3：多宠物用户
            createBehavior("user003", "buy", "狗狗", "狗零食", "狗粮", System.currentTimeMillis() - 86400000L),
            createBehavior("user003", "buy", "猫咪", "猫主粮", "猫粮", System.currentTimeMillis() - 172800000L),
            createBehavior("user003", "cart", "仓鼠类及其它小宠", "仓鼠", "仓鼠粮", System.currentTimeMillis() - 259200000L),
            
            // 用户4：鸟类用户
            createBehavior("user004", "buy", "鸟类及用品", "鸟粮", "鸟粮", System.currentTimeMillis() - 86400000L),
            createBehavior("user004", "favorite", "鸟类及用品", "鸟笼", "鸟笼", System.currentTimeMillis() - 172800000L)
        );
        
        return env.fromCollection(testData);
    }
    
    private static UserBehavior createBehavior(String userId, String behaviorType, 
                                             String cat1, String cat2, String cat3, 
                                             Long timestamp) {
        UserBehavior behavior = new UserBehavior();
        behavior.setUserId(userId);
        behavior.setBehaviorType(behaviorType);
        behavior.setCategory1(cat1);
        behavior.setCategory2(cat2);
        behavior.setCategory3(cat3);
        behavior.setTimestamp(timestamp);
        
        Map<String, String> attributes = new HashMap<>();
        behavior.setItemAttributes(attributes);
        
        return behavior;
    }
}